package com.navinfo.platform.dataanalysis.service.impl;

import com.mongodb.spark.rdd.api.java.JavaMongoRDD;
import com.navinfo.platform.common.dto.DataInfo;
import com.navinfo.platform.common.service.ICommonService;
import com.navinfo.platform.common.utils.TidPartitioner;
import com.navinfo.platform.dataanalysis.dto.*;
import com.navinfo.platform.dataanalysis.dto.condition.*;
import com.navinfo.platform.dataanalysis.service.ILocationDataBussinessService;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.*;
import org.bson.Document;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.util.*;
import java.util.stream.Collectors;

@SuppressWarnings("Duplicates")
public enum LocationDataBusinessService implements ILocationDataBussinessService {

    INSTANCE;
    private static Logger logger = LoggerFactory.getLogger(LocationDataBusinessService.class);
    @Override
    public JavaPairRDD<Long, List<DataInfo>> convert(Dataset<Row> realtimeDataset, Map<String, String> configMap) {
        int partitionNum = getPartitionNum(configMap);
        return realtimeDataset.map(RealtimeDataDto::parseRow, Encoders.bean(DataInfo.class))
                .toJavaRDD().mapPartitionsToPair(LocationDataBusinessService.realtimeMapToPair())
                .partitionBy(new TidPartitioner(partitionNum))
                .reduceByKey((list, list2) -> {
                    list.addAll(list2);
                    return list;
                }).mapPartitionsToPair(d->{
                    List<Tuple2<Long, List<DataInfo>>> retList = new ArrayList<>();
                    while(d.hasNext()){
                        Tuple2<Long, List<DataInfo>> tuple2 = d.next();
                        long tid = tuple2._1();
                        List<DataInfo> realtimeDataDtos = tuple2._2();
                        realtimeDataDtos.sort(Comparator.comparingLong(DataInfo::getGpsDate));
                        retList.add(Tuple2.apply(tid, realtimeDataDtos));
                    }
                    return retList.iterator();
                });
    }

    /**
     * 获取分区数
     * @param configMap
     * @return
     */
    private int getPartitionNum(Map<String, String> configMap) {
        String partitionNumStr = configMap.get(ICommonService.PARTITION_NUM_VALUE);
        int partitionNum;
        if(partitionNumStr == null){
            partitionNum = 500;
        } else {
            partitionNum = Integer.parseInt(partitionNumStr);
        }
        return partitionNum;
    }

    private void statisticFaultInfo(JavaMongoRDD<Document> javaMongoRDD, Map<String, String> configMap) {
        int partitionNum = getPartitionNum(configMap);
        JavaRDD<FaultInfoDto> faultInfoRDD = javaMongoRDD.map(FaultInfoDto::parseRow);
        faultInfoRDD.mapPartitionsToPair(LocationDataBusinessService.faultInfoToPair()).partitionBy(new TidPartitioner(partitionNum))
                .reduceByKey((list, list2) -> {
                    list.addAll(list2);
                    return list;
                }).mapPartitions(f ->{
            Map<Long, Map<String, Integer>> faultTimes = new HashMap<>();
            while(f.hasNext()){
                Tuple2<Long, List<FaultInfoDto>> tuple2 = f.next();
                long tid = tuple2._1();
                List<FaultInfoDto> faultInfoDtos = tuple2._2();
                faultTimes.computeIfAbsent(tid, d -> new HashMap<>());
                Map<String, Integer> innerFaultTimes = faultTimes.get(tid);
                int allTimes = 0;
                for(FaultInfoDto faultInfoDto : faultInfoDtos){
                    int spn = faultInfoDto.getSpn();
                    int fmi = faultInfoDto.getFmi();
                    String key = spn + "_" + fmi;
                    String allKey = "-1";
                    innerFaultTimes.putIfAbsent(key, 0);
                    int times = innerFaultTimes.get(key);
                    innerFaultTimes.put(key, times + 1);
                    innerFaultTimes.put(allKey, allTimes);
                }
            }
            return null;
        });
    }



    @Override
    public JavaPairRDD<Long, List<DataInfo>> unionData(JavaPairRDD<Long, List<DataInfo>> locationDataRdd, JavaPairRDD<Long, List<DataInfo>> realtimeDataRdd) {
        JavaPairRDD<Long, Long> locationRDD = locationDataRdd.mapToPair(f -> Tuple2.apply(f._1(), f._1()));
        JavaPairRDD<Long, Long> realtimeRDD = realtimeDataRdd.mapToPair(f -> Tuple2.apply(f._1(), f._1()));
        return locationRDD.subtract(realtimeRDD).join(locationDataRdd).mapToPair(Tuple2::_2).union(realtimeDataRdd);
    }

    @Override
    public void statisticInfo(SparkSession spark, JavaPairRDD<Long, List<DataInfo>> locationPairRDD, JavaPairRDD<Long, List<DataInfo>> dataInfoRdd, Dataset<Row> basicInfo,
                              Dataset<Row> idleInfo, JavaMongoRDD<Document> javaMongoRDD, Map<String, String> config) {
        //1 .统计事件
        //熄火
        JavaRDD<FoutInfoDto> foutInfoDtoJavaRDD = statisticFout(locationPairRDD, config);
        List<FoutInfoDto> foutInfoDtos = foutInfoDtoJavaRDD.collect();
        Map<Long, FoutInfoDto> foutInfoData = foutInfoDtos.stream().collect(Collectors.toMap(FoutInfoDto::getTid, v -> v));
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
        final Broadcast<Map<Long, FoutInfoDto>> foutInfoDtoMapBroadcast = jsc.broadcast(foutInfoData);
        final Broadcast<Map<String, String>> configMapBroadcast = jsc.broadcast(config);
        JavaRDD<StatisticsInfoDto> statisticsInfoDtoJavaRDD = dataInfoRdd.mapPartitions(f -> {
            Map<String, String> configMap = configMapBroadcast.getValue();
            String day = configMap.get(ICommonService.DAY_VALUE);
            Map<Long, EventCache> csbInfoDtoMap = new HashMap<>();
            Map<Long, EventCache> hsbInfoDtoMap = new HashMap<>();
            Map<Long, EventCache> clutchInfoDtoMap = new HashMap<>();
            Map<Long, EventCache> ngInfoDtoMap = new HashMap<>();
            Map<Long, EventCache> hs90kmInfoDtoMap = new HashMap<>();
            Map<Long, EventCache> hs120kmInfoDtoMap = new HashMap<>();
            Map<Long, EventCache> srInfoDtoMap = new HashMap<>();
            Map<Long, EventCache> lrogInfoDtoMap = new HashMap<>();

            CsbCondition csbCondition = new CsbCondition();
            HsbCondition hsbCondition = new HsbCondition();
            ClutchCondition clutchCondition = new ClutchCondition();
            NgCondition ngCondition = new NgCondition();
            Hs90kmCondition hs90kmCondition = new Hs90kmCondition();
            Hs120kmCondition hs120kmCondition = new Hs120kmCondition();
            SrCondition srCondition = new SrCondition();
            LrCondition lrCondition = new LrCondition();

            List<StatisticsInfoDto> statisticsInfoDtos = new ArrayList<>();
            Map<Long, FoutInfoDto> foutInfoDtoMap = foutInfoDtoMapBroadcast.getValue();
            String[] csbParam = configMap.get("csbCondition").split(",");
            String[] hsbParam = configMap.get("hsbCondition").split(",");
            String[] ngParam = configMap.get("ngCondition").split(",");
            String[] hs90Param = configMap.get("hs90Condition").split(",");
            String[] hs120Param = configMap.get("hs120Condition").split(",");
            String[] srParam = configMap.get("srParam").split(",");
            String[] lrogParam = configMap.get("lrogParam").split(",");

            while(f.hasNext()){
                Tuple2<Long, List<DataInfo>> tuple2 = f.next();
                long tid = tuple2._1();
                List<DataInfo> dataInfos = tuple2._2();
                initialization(csbInfoDtoMap, hsbInfoDtoMap, clutchInfoDtoMap, ngInfoDtoMap, hs90kmInfoDtoMap, hs120kmInfoDtoMap, srInfoDtoMap, lrogInfoDtoMap, tid);
                EventCache csbInfoCache = csbInfoDtoMap.get(tid);
                EventCache hsbInfoCache = hsbInfoDtoMap.get(tid);
                EventCache clutchInfoCache = clutchInfoDtoMap.get(tid);
                EventCache ngInfoCache = ngInfoDtoMap.get(tid);
                EventCache hs90kmInfoCache = hs90kmInfoDtoMap.get(tid);
                EventCache hs120kmInfoCache = hs120kmInfoDtoMap.get(tid);
                EventCache srInfoCache = srInfoDtoMap.get(tid);
                EventCache lrogInfoCache = lrogInfoDtoMap.get(tid);

                int count = 1;
                int size = dataInfos.size();
                boolean lastPoint = false;
                StatisticsInfoDto statisticsInfoDto = new StatisticsInfoDto();
                statisticsInfoDto.setTid(tid);
                statisticsInfoDto.setDay(day);

                for(DataInfo dataInfo : dataInfos){
                    if(count == size){
                        lastPoint = true;
                    }

                    //event_info
                    calcEventInfo(csbInfoCache, dataInfo, csbCondition, lastPoint, csbParam);
                    calcEventInfo(hsbInfoCache, dataInfo, hsbCondition, lastPoint, hsbParam);
                    calcEventInfo(clutchInfoCache, dataInfo, clutchCondition, lastPoint);
                    calcEventInfo(ngInfoCache, dataInfo, ngCondition, lastPoint, ngParam);
                    //alarm_info
                    calcEventInfo(hs90kmInfoCache, dataInfo, hs90kmCondition, lastPoint, hs90Param);
                    calcEventInfo(hs120kmInfoCache, dataInfo, hs120kmCondition, lastPoint, hs120Param);
                    calcEventInfo(srInfoCache, dataInfo, srCondition, lastPoint, srParam);
                    calcEventInfo(lrogInfoCache, dataInfo, lrCondition, lastPoint, lrogParam);
                    count++;
                }
                setStatisticsInfo(foutInfoDtoMap, tid, csbInfoCache, hsbInfoCache, clutchInfoCache, ngInfoCache, hs90kmInfoCache,
                        hs120kmInfoCache, srInfoCache, lrogInfoCache, statisticsInfoDto);

                //添加数据到集合中
                statisticsInfoDtos.add(statisticsInfoDto);
            }
            return statisticsInfoDtos.iterator();
        });

        Dataset<Row> mainstatisticsInfo = spark.createDataFrame(statisticsInfoDtoJavaRDD, StatisticsInfoDto.class);
        Dataset<Row> eventInfoDataset = mainstatisticsInfo.join(basicInfo, "tid").join(idleInfo, "tid");
        Dataset<StatisticsInfoDto> saveData = eventInfoDataset.map(StatisticsInfoDto::parseRow, Encoders.bean(StatisticsInfoDto.class));
        saveData.persist();
        saveEventInfo(config, saveData);
        saveAlarmInfo(config, saveData);
//        statisticFaultInfo(javaMongoRDD);

    }

    private void setStatisticsInfo(Map<Long, FoutInfoDto> foutInfoDtoMap, long tid, EventCache csbInfoCache, EventCache hsbInfoCache, EventCache clutchInfoCache, EventCache ngInfoCache, EventCache hs90kmInfoCache, EventCache hs120kmInfoCache, EventCache srInfoCache, EventCache lrogInfoCache, StatisticsInfoDto statisticsInfoDto) {
        //设置普通刹车统计数据
        statisticsInfoDto.setCsbTimes(csbInfoCache.getTimes());
        statisticsInfoDto.setCsbDuration(csbInfoCache.getDuration());
        statisticsInfoDto.setCsbMileage(csbInfoCache.getMileage());
        //设置高速刹车统计数据
        statisticsInfoDto.setHsbTimes(hsbInfoCache.getTimes());
        statisticsInfoDto.setHsbDuration(hsbInfoCache.getDuration());
        statisticsInfoDto.setHsbMileage(hsbInfoCache.getMileage());
        //设置离合统计数据
        statisticsInfoDto.setClutchTimes(clutchInfoCache.getTimes());
        statisticsInfoDto.setClutchDuration(clutchInfoCache.getDuration());
        statisticsInfoDto.setClutchMileage(clutchInfoCache.getMileage());
        //设置空挡滑行统计
        statisticsInfoDto.setNgTimes(ngInfoCache.getTimes());
        statisticsInfoDto.setNgDuration(ngInfoCache.getDuration());
        statisticsInfoDto.setNgMileage(ngInfoCache.getMileage());

        FoutInfoDto foutInfoDto = foutInfoDtoMap.get(tid);
        if(foutInfoDto != null){
            statisticsInfoDto.setFoutTimes(foutInfoDto.getFoutTimes());
            statisticsInfoDto.setFoutDuration(foutInfoDto.getFoutDuration());
        }
        //超速90km数据hs90kmInfoCache
        statisticsInfoDto.setSo90Times(hs90kmInfoCache.getTimes());
        statisticsInfoDto.setSo90Duration(hs90kmInfoCache.getDuration());
        statisticsInfoDto.setSo90Mileage(hs90kmInfoCache.getMileage());

        //超速120km数据
        statisticsInfoDto.setSo120Times(hs120kmInfoCache.getTimes());
        statisticsInfoDto.setSo120Duration(hs120kmInfoCache.getDuration());
        statisticsInfoDto.setSo120Mileage(hs120kmInfoCache.getMileage());

        //超转速
        statisticsInfoDto.setSrTimes(srInfoCache.getTimes());
        statisticsInfoDto.setSrDuration(srInfoCache.getDuration());
        statisticsInfoDto.setSrMileage(srInfoCache.getMileage());
        //低转速
        statisticsInfoDto.setLrogTimes(lrogInfoCache.getTimes());
        statisticsInfoDto.setLrogDuration(lrogInfoCache.getDuration());
        statisticsInfoDto.setLrogMileage(lrogInfoCache.getMileage());
    }

    private void initialization(Map<Long, EventCache> csbInfoDtoMap, Map<Long, EventCache> hsbInfoDtoMap, Map<Long, EventCache> clutchInfoDtoMap, Map<Long, EventCache> ngInfoDtoMap, Map<Long, EventCache> hs90kmInfoDtoMap, Map<Long, EventCache> hs120kmInfoDtoMap, Map<Long, EventCache> srInfoDtoMap, Map<Long, EventCache> lrogInfoDtoMap, long tid) {
        csbInfoDtoMap.computeIfAbsent(tid, d -> new EventCache());
        hsbInfoDtoMap.computeIfAbsent(tid, d -> new EventCache());
        clutchInfoDtoMap.computeIfAbsent(tid, d -> new EventCache());
        ngInfoDtoMap.computeIfAbsent(tid, d -> new EventCache());
        hs90kmInfoDtoMap.computeIfAbsent(tid, d -> new EventCache());
        hs120kmInfoDtoMap.computeIfAbsent(tid, d -> new EventCache());
        srInfoDtoMap.computeIfAbsent(tid, d -> new EventCache());
        lrogInfoDtoMap.computeIfAbsent(tid, d -> new EventCache());
    }


    private void saveEventInfo(Map<String, String> config, Dataset<StatisticsInfoDto> saveData) {
        String pathPrefix = config.get("hive.event.path");
        String day = config.get(ICommonService.DAY_VALUE);
        String path = String.format(pathPrefix, day);
        saveData.select(StatisticsInfoDto.eventColumns()).repartition(1).write().mode(SaveMode.Overwrite).parquet(path);
    }

    private void saveAlarmInfo(Map<String, String> config, Dataset<StatisticsInfoDto> saveData) {
        String pathPrefix = config.get("hive.alarm.path");
        String day = config.get(ICommonService.DAY_VALUE);
        String path = String.format(pathPrefix, day);
        saveData.select(StatisticsInfoDto.alarmColumns()).repartition(1).write().mode(SaveMode.Overwrite).parquet(path);
    }

    public JavaRDD<FoutInfoDto> statisticFout(JavaPairRDD<Long, List<DataInfo>> dataInfoRdd, Map<String, String> configMap) {
         return dataInfoRdd.mapPartitions(d -> {
            Map<Long, EventCache> foutInfoDtoMap = new HashMap<>();
            FoutCondition foutCondition = new FoutCondition();
            List<FoutInfoDto> foutInfoDtos = new ArrayList<>();
            while(d.hasNext()){
                Tuple2<Long, List<DataInfo>> tuple2 = d.next();
                long tid = tuple2._1();
                List<DataInfo> dataInfos = tuple2._2();
                foutInfoDtoMap.computeIfAbsent(tid, f -> new EventCache());
                EventCache foutInfoCache = foutInfoDtoMap.get(tid);
                FoutInfoDto foutInfoDto = new FoutInfoDto();
                int count = 1;
                int size = dataInfos.size();
                boolean lastPoint = false;

                for(DataInfo dataInfo : dataInfos){
                    if(count == size){
                        lastPoint = true;
                    }
                    calcEventInfo(foutInfoCache, dataInfo, foutCondition, lastPoint);
                    count++;
                }
                foutInfoDto.setTid(tid);
                foutInfoDto.setFoutTimes(foutInfoCache.getTimes());
                foutInfoDto.setFoutDuration(foutInfoCache.getDuration());
                foutInfoDtos.add(foutInfoDto);
            }
            return foutInfoDtos.iterator();
        });
    }

    public void calcEventInfo(EventCache eventCache, DataInfo dataInfo, AbstractCondition conditionBuilder, boolean lastPoint){
        calcEventInfo(eventCache, dataInfo, conditionBuilder, lastPoint, null);
    }

    public void calcEventInfo(EventCache eventCache, DataInfo dataInfo, AbstractCondition conditionBuilder, boolean lastPoint, String[] param) {
        int speed = dataInfo.getSpeed();
        int gpsDate = (int)dataInfo.getGpsDate();

        conditionBuilder.setParam(param);
        conditionBuilder.setLastPoint(lastPoint);
        eventCache.setCurrDate(gpsDate);
        int diff = eventCache.getDiffDuration();
        boolean flag = conditionBuilder.condition(dataInfo, diff);

        if(!eventCache.isLastStatus() && flag){
            eventCache.setGpsDate(gpsDate);
        }
        int oriDuration = eventCache.calcDuration(200);
        int duration = flag ? oriDuration : 0;
        int integralMileage = speed * duration;
        if(lastPoint){
            if(eventCache.isLastStatus()){
                eventCache.incrMileage(integralMileage);
                eventCache.incrDuration(duration);
                eventCache.incrTimes();
                if(conditionBuilder instanceof NgCondition){
                    if(eventCache.getDuration() < 60){
                        eventCache.incrMileage(-integralMileage);
                        eventCache.incrDuration(-duration);
                        eventCache.incrTimes(-1);
                    }
                }
                if(conditionBuilder instanceof FoutCondition) {
                    eventCache.incrMileage(-integralMileage);
                }
            }
        } else {
            if(flag){
                eventCache.incrDuration(duration);
                eventCache.incrMileage(integralMileage);
                if(conditionBuilder instanceof NgCondition){
                    if(duration < 60){
                        eventCache.incrDuration(-duration);
                        eventCache.incrMileage(-integralMileage);
                    }
                }
                if(conditionBuilder instanceof FoutCondition){
                    eventCache.incrMileage(-integralMileage);
                }
            } else {
                if(eventCache.isLastStatus() && !flag){
                    eventCache.incrTimes();
                    if(conditionBuilder instanceof NgCondition){
                        if(eventCache.getDuration() < 60){
                            eventCache.incrTimes(-1);
                        }
                    }
                }
            }
        }
//        logger.error(String.format("tid:%s, lastPoint:%s, lastStatus:%s, lastGpsDate:%s, currGpsDate:%s, flag:%s, times:%s, diff:%s, oriDuration:%s, duration:%s, allDuration:%s, integralMileage:%s",
//                dataInfo.getTid(), lastPoint, eventCache.isLastStatus(), eventCache.getGpsDate(), eventCache.getCurrDate(), flag, eventCache.getTimes(), diff, oriDuration, duration, eventCache.getDuration(), integralMileage));
        eventCache.setGpsDate(gpsDate);
        eventCache.setLastStatus(flag);
    }

    private static PairFlatMapFunction<Iterator<DataInfo>, Long, List<DataInfo>>  realtimeMapToPair() {
        return d -> {
            List<DataInfo> dataList = new ArrayList<>();
            d.forEachRemaining(dataList::add);
            return dataList.stream().collect(Collectors.groupingBy(DataInfo::getTid))
                    .entrySet().stream().map(entry -> new Tuple2<>(entry.getKey(), entry.getValue())
                    ).collect(Collectors.toList()).iterator();
        };
    }

    private static PairFlatMapFunction<Iterator<FaultInfoDto>, Long, List<FaultInfoDto>> faultInfoToPair() {
        return d -> {
            List<FaultInfoDto> dataList = new ArrayList<>();
            d.forEachRemaining(dataList::add);
            return dataList.stream().collect(Collectors.groupingBy(FaultInfoDto::getTid))
                    .entrySet().stream().map(entry -> new Tuple2<>(entry.getKey(), entry.getValue())
                    ).collect(Collectors.toList()).iterator();
        };
    }
}
